package com.study;

import com.study.entity.UserBehavior;
import com.study.utils.JsonHelper;
import com.typesafe.sslconfig.ssl.FakeChainedKeyStore;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;

import java.util.Date;
import java.util.Properties;

/**
 * @author sjw
 * @className SoureKafka
 * @date 2020/11/20 8:13
 * @description: TODO
 */
public class SoureKafkaStreamWindow {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        Properties props=new Properties();
        props.put("bootstrap.servers", "192.168.10.200:9092");
        props.put("zookeeper.connect", "192.168.10.200:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //key 反序列化
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest"); //value 反序列化

        DataStreamSource<String> dataStreamSource=env.addSource(new FlinkKafkaConsumer010<String>(
                "logEvent",new SimpleStringSchema(),props)).setParallelism(1);
        SingleOutputStreamOperator<UserBehavior> userBehaviorSingleOutputStreamOperator= dataStreamSource.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                return JsonHelper.jsonToObject(s,UserBehavior.class);
            }
        });

        KeyedStream<UserBehavior, Tuple> keyedStream= userBehaviorSingleOutputStreamOperator.keyBy("userId");

        //WindowedStream<UserBehavior, Tuple, TimeWindow>   timeWindowedStream= keyedStream.timeWindow(Time.seconds(10));
        //timeWindowedStream.apply(new TimeWindowFunction()).print();


        WindowedStream<UserBehavior, Tuple, GlobalWindow>   countWindowedStream= keyedStream.countWindow(10);
        countWindowedStream.apply(new CountWindowFunction()).print();



        env.execute("Flink add data source kafak ");
    }


    private static class TimeWindowFunction implements WindowFunction<UserBehavior, Object, Tuple, TimeWindow>{

        @Override
        public void apply(Tuple tuple, TimeWindow window, Iterable<UserBehavior> input, Collector<Object> out) throws Exception {
            int i=0;
            for(UserBehavior userBehavior: input){
                i++;
                System.out.println(tuple.getField(0)+"    "+userBehavior.userId+"    "+userBehavior.getBehavior()+"  "+new Date(userBehavior.getTimestamp()));
            }
            out.collect("key "+tuple.getField(0)+"  sum   "+i);
        }
    }


    private static class CountWindowFunction implements WindowFunction<UserBehavior, Object, Tuple, GlobalWindow>{

        @Override
        public void apply(Tuple tuple, GlobalWindow window, Iterable<UserBehavior> input, Collector<Object> out) throws Exception {
            int i=0;
            for(UserBehavior userBehavior: input){
                i++;
                System.out.println(tuple.getField(0)+"    "+userBehavior.userId+"    "+userBehavior.getBehavior()+"  "+new Date(userBehavior.getTimestamp()));
            }
            out.collect("key "+tuple.getField(0)+"  sum   "+i);
        }
    }
}


